package com.example.kafkademo.kafkatest1;


import com.example.kafkademo.domian.InfoMsg;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class ProducerA  {

    public static String topic = "test";//定义主题

    public static void main(String[] args) throws InterruptedException {
        Properties p = new Properties();
        p.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.4.43:9092");//kafka地址，多个地址用逗号分割
        p.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);// key
        p.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);//value
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(p);

        try {
            while (true) {
//                String msg = "Hello," + new Random().nextInt(100);
                InfoMsg msg = new InfoMsg();

                msg.setFphone_no("13800138000");
                msg.setFsystem_type("ios");
                msg.setFvisit_area_code("13456");
                msg.setFlac("lac9980");
                msg.setFimsi("qwe34er3ew3se898");
                msg.setFimei("123456789op");
                msg.setFcell_id("45689");
                msg.setFstart_time("2031-10-29");
                msg.setFhome_area_code("家的区域类型");
                msg.setFother_party("其他数据");
                msg.setFcall_type("设备运作类型");

                ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic,msg.toString());
                kafkaProducer.send(record);
                System.out.println("消息发送成功:" + msg);
                Thread.sleep(500);
            }
        } finally {
            kafkaProducer.close();
        }

    }
}
